package org.fastsyncer.manager.log;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Pipeline {

    private static final Logger logger = LoggerFactory.getLogger(Pipeline.class);
   
    private LogFactory logFactory;
    
    // 驱动ID
    private final String id;

    // 新增日志
    private final BlockingQueue<StringBuilder> newLog;
    
    // 删除日志
    private final List<Row> delLog;

    // 文件锁
    private final ReentrantLock lock;

    public Pipeline(LogFactory logFactory, String id) {
        super();
        this.logFactory = logFactory;
        this.id = id;
        lock = new ReentrantLock();
        newLog = new LinkedBlockingQueue<>();
        delLog = new LinkedList<>();
    }

    public void writeLog(StringBuilder row) {
        newLog.offer(row);
    }
    
    public void delLog(Row r) {
        delLog.add(r);
    }

    public void flush() {
        boolean tryLock = false;
        try {
            // 10s之内获取锁，否则失败
            tryLock = lock.tryLock(10, TimeUnit.SECONDS);
            
            // 1.写入
            StringBuilder rows = new StringBuilder();
            while (!newLog.isEmpty()) {
                rows.append(newLog.poll());
            }
            logFactory.writeLog(id, rows, true);
            
            // 2.删除
            logFactory.deleteLog(id, delLog);
        } catch (InterruptedException e) {
            logger.error("Acquire file lock failed while write the driver ID \"%s\".", id);
        } finally {
            if (tryLock) {
                lock.unlock();
            }
        }
    }
    
}